package org.weishe.rtr.mr;

import java.io.IOException;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.util.StringUtils;

/**
 * 数据清洗 将文件中的用ip表示的删选出来换成经纬度
 * 
 * @author chenbiao
 *
 */
public class Step1 {
	public static class S1Mapper extends Mapper<LongWritable, Text, Text, NullWritable> {
		private NullWritable nw = NullWritable.get();
		private String out = "";
		private Text outv = new Text();

		@Override
		protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, NullWritable>.Context context)
				throws IOException, InterruptedException {
			String str[] = value.toString().split("\t");
			if (str != null || str.length > 5) {
				if (str.length == 6) {
					context.write(value, nw);
				} else if (str.length == 5) {
					try {
						String ip[] = StringUtils.split(str[2], '.');
						out = str[0] + "\t" + str[1] + "\t" + str[3] + "\t" + str[4] + "\t" + ip[2] + "\t" + ip[3];
						outv.set(out);
						context.write(outv, nw);
					} catch (Exception e) {

						outv.set("error:" + value.toString() + "   ip:" + str[2]);
						context.write(outv, nw);

						System.out.println(value.toString());
						System.out.println("ip:" + str[2]);
						e.printStackTrace();
					}
				}
			}
		}

	}
}
